package com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.utils

import org.apache.spark.sql.{DataFrame, SparkSession}

class newAdd_brand_util {

  /**
    * @ describe 新增品牌提取
    * source_data：原始数据
    * old_brand_data老品牌表
    */
  def new_brand_excat(source_data:DataFrame,old_brand_data:DataFrame,spark:SparkSession,timeStamp:String):DataFrame = {
//    从原始数据中提取字段：platformName,brandValueId,brandName,firstCategoryId
    val sourcedata = source_data.selectExpr(s"platformName", "brandValueId", "brandName", "firstCategoryId")
      .registerTempTable("sourcedata")


//    品牌表路径
    old_brand_data.registerTempTable("brandtable")

    val brandadd = spark.sqlContext.sql(
      s"""
        |select
        |'${timeStamp}'timeStamp,
        |t1.brandCcId,
        |t1.platformName,
        |t1.brandValueId,
        |t1.brandName,
        |t1.firstCategoryId
        |from
        |(select *,
        |concat(brandValueId,"|",firstCategoryId) brandCcId
        |from sourcedata
        |) t1
        |left join
        |brandtable t2
        |on t1.brandCcId = t2.brandCcId
        |where t2.brandCcId is null
      """.stripMargin)
//      将没有的字段补全
        .na.fill(Map("brandCcId"->"-1"))
      .dropDuplicates("brandCcId")
      .where("brandName != '-1' and brandValueId != '-1' ")


    return brandadd
  }
}
